﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AutoMapper;
using Confluent.Kafka;
 

 
using Microsoft.AspNetCore.Mvc;
using Microsoft.EntityFrameworkCore;
using MySql.Data.MySqlClient;
using Newtonsoft.Json;
using Spire.Presentation;

using Ccteg.Common;
using Ccteg.DB;

namespace Cctegaq.Api.Controllers.Api.V1.Rbac
{
    /// <summary>
    /// Kafak消息发送
    /// </summary>
    [ApiController]
    [Route("api/v1/rbac/[controller]/[action]")]
    public class KafakController : Controller
    {
        private readonly MySqlDbContext _dbContext;
        private readonly IMapper _mapper;
        /// <summary>
        /// 
        /// </summary>
        /// <param name="dbContext"></param>
        /// <param name="mapper"></param>
        public KafakController(MySqlDbContext dbContext, IMapper mapper)
        {
            _dbContext = dbContext;
            _mapper = mapper;
        }


        private void CallBack(object data)
        {
            int i = 0;
        }


        /// <summary>
        /// 生产（同步发送） 测试自己封装的 回头优化异步
        /// </summary>
        /// <param name="topic">主题名称</param>
        /// <param name="jsondata">内容 对应他们的data</param>
        /// <returns></returns>
        //[HttpPost]
        //public IActionResult kafkatest_Producer([FromForm] string topic, [FromForm] string jsondata)
        //{
        //    ISUPKafkaProducer<string, string> kafkaCenter = SUPKafkaMessageCenter<string, string>.CreateTopicProducer("61.132.89.53:3090");
        //    kafkaCenter.Send("kafkatest2", jsondata, topic, deliveryReport =>
        //    {
        //        int bb = 1000;

        //    });
        //    return Ok(1);
        //}

        /// <summary>
        /// 生产
        /// </summary>
        /// <param name="topic">主题名称</param>
        /// <param name="jsondata">内容 对应他们的data</param>
        /// <returns></returns>


        [HttpPost]
        public ResponseModel<string> Producer([FromForm] string topic, [FromForm] string jsondata)
        {
            var response = new ResponseModel<string>();
            response.SetFailed(); //默认为失败
            try
            {

                var pConfig = new ProducerConfig
                {
                    BootstrapServers = "61.132.89.53:3090",

                };

                using (var producer = new ProducerBuilder<Null, string>(pConfig).Build())
                {


                    Task.Run(async () =>
                    {
                        var task = await producer.ProduceAsync(topic, new Message<Null, string> { Value = jsondata });

                        response.SetSuccess();
                        producer.Flush(TimeSpan.FromSeconds(5));

                    }).Wait();


                }

            }
            catch (Exception ex)
            {
                response.SetFailed(ex.Message);

            }
            return response;

            //var deliveryReport = producer.ProduceAsync(topic, new Message<Null, string> { Value = jsondata }, new CancellationTokenSource(5000).Token);
            //var deliveryReport = producer.ProduceAsync(topic, new Message<Null, string> { Value = jsondata });

            //if (task.Status == TaskStatus.Faulted)
            //    response.SetFailed();
            //else
            //    response.SetSuccess();
            //deliveryReport.ContinueWith(task =>
            //{
            //    response.Data = "Producer：" + producer.Name + "\r\nTopic：" + topic + "\r\nPartition：" + task.Result.Partition + "\r\nOffset：" + task.Result.Offset + "\r\nMessage：" + task.Result.Value + "\r\n" + "taskstatus=" + (int)task.Status;
            //    if (task.Status == TaskStatus.Faulted)
            //        response.SetFailed();
            //    else
            //        response.SetSuccess();
            //});

        }

        /// <summary>
        /// 消费 每次消费一条
        /// </summary>
        /// <param name="topic">你要消费的主题</param>
        /// <returns></returns>
        [HttpPost]
        public IActionResult Consumer(string topic)
        {
            var response = new ResponseModel<string>();
            response.SetFailed(); //默认为失败
            var conf = new ConsumerConfig
            {
                GroupId = "test-consumer-group2",
                BootstrapServers = "61.132.89.53:3090",
                AutoOffsetReset = AutoOffsetReset.Latest,
                EnableAutoCommit = false  // 设置非自动偏移，业务逻辑完成后手动处理偏移，防止数据丢失
            };

            using (var consumer = new ConsumerBuilder<Null, string>(conf).Build())
            {
                // 订阅topic
                consumer.Subscribe(topic);

                while (true)
                {

                    try
                    {

                        var consumeResult = consumer.Consume(TimeSpan.FromSeconds(6));
                        // 开始我的业务逻辑
                        response.SetSuccess();
                        consumer.Commit(consumeResult); //从队列中移除
                        response.SetData(consumeResult.Message.Value);
                        response.Message = "我消费到了信息";
                        return Ok(response);

                    }
                    catch (Exception e)
                    {
                        response.SetFailed(e.Message);
                        return Ok(response);


                    }

                }
            }
        }


    }
}
